package com.songyang.yougnsql.core.version;

import com.songyang.yougnsql.config.Config;
import com.songyang.yougnsql.config.ConfigConst;
import com.songyang.yougnsql.core.cache.AbstractCache;
import com.songyang.yougnsql.common.ErrorMsg;
import com.songyang.yougnsql.common.YoungSQLException;
import com.songyang.yougnsql.core.data.DataManager;
import com.songyang.yougnsql.core.transaction.TransactionManager;
import com.songyang.yougnsql.core.transaction.TransactionManagerImpl;

import java.io.ObjectInputFilter;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @ClassName VersionManagerImpl
 * @Description
 * @date 2022/12/16 10:38
 * @Author yanceysong
 * @Version 1.0
 */
public class VersionManagerImpl extends AbstractCache<VersionEntry> implements VersionManager {
    private TransactionManager transactionManager;
    private DataManager dataManager;
    private Map<Long, Transaction> activeTransaction;
    private Lock lock;
    private LockMap lockMap;

    public VersionManagerImpl(int maxResource) {
        super(maxResource);
    }

    public VersionManagerImpl(TransactionManager tm, DataManager dm) {
        super(Integer.parseInt(String.valueOf((Integer) Config.getConfig(ConfigConst.CORE_DATA_CACHE_LIMIT))));
        this.transactionManager = tm;
        this.dataManager = dm;
        this.activeTransaction = new HashMap<>();
        activeTransaction.put(TransactionManager.SUPER_XID, Transaction.newTransaction(TransactionManagerImpl.SUPER_XID, 0, null));
        this.lock = new ReentrantLock();
        this.lockMap = new LockMap();
    }

    @Override
    public byte[] read(long xid, long uid) throws Exception {
        lock.lock();
        Transaction transaction = activeTransaction.get(xid);
        lock.unlock();
        VersionEntry entry = null;
        try {
            //找到uid的entry
            entry = super.get(uid);
            if (Visibility.isVisible(transactionManager, transaction, entry)) {
                return entry.data();
            } else return null;
        } catch (Exception e) {
            if (e.getMessage().equals(ErrorMsg.NULL_ENTRY_EXCEPTION)) return null;
            throw e;
        } finally {
            assert entry != null;
            entry.release();
        }
    }

    @Override
    public long insert(long xid, byte[] data) throws Exception {
        lock.lock();
        Transaction t = activeTransaction.get(xid);
        lock.unlock();
        byte[] raw = VersionEntry.wrapEntryRaw(xid, data);
        return dataManager.insert(xid, raw);
    }

    @Override
    public boolean delete(long xid, long uid) throws Exception {
        lock.lock();
        Transaction t = activeTransaction.get(xid);
        lock.unlock();
        VersionEntry entry = null;
        try {
            entry = super.get(uid);
        } catch (Exception e) {
            if (e.getMessage().equals(ErrorMsg.NULL_ENTRY_EXCEPTION)) {
                return false;
            } else {
                throw e;
            }
        }
        try {
            if (!Visibility.isVisible(transactionManager, t, entry)) {
                return false;
            }
            Lock l = null;
            try {
                l = lockMap.add(xid, uid);
            } catch (Exception e) {
                internAbort(xid, true);
                t.setAutoAborted(true);
                throw new YoungSQLException(ErrorMsg.CONCURRENT_UPDATE_EXCEPTION);
            }
            if (l != null) {
                l.lock();
                l.unlock();
            }
            if (entry.getXMax() == xid) {
                return false;
            }
            if (Visibility.isVersionSkip(transactionManager, t, entry)) {
                internAbort(xid, true);
                t.setAutoAborted(true);
                throw new YoungSQLException(ErrorMsg.CONCURRENT_UPDATE_EXCEPTION);
            }
            entry.setXMax(xid);
            return true;
        } finally {
            entry.release();
        }
    }

    /**
     * 开启一个事务
     *
     * @param level 隔离级别
     * @return 事务的id
     */
    @Override
    public long begin(int level) {
        lock.lock();
        try {
            long xid = transactionManager.begin();
            activeTransaction.put(xid, Transaction.newTransaction(xid, level, activeTransaction));
            return xid;
        } finally {
            lock.unlock();
        }
    }

    @Override
    public void commit(long xid) {
        lock.lock();
        try {
            Transaction transaction = activeTransaction.get(xid);
            activeTransaction.remove(xid);
            lockMap.remove(xid);
            transactionManager.commit(xid);
        } finally {
            lock.unlock();
        }
    }

    @Override
    public void abort(long xid) {
        internAbort(xid, false);
    }

    private void internAbort(long xid, boolean autoAbort) {
        lock.lock();
        Transaction transaction = activeTransaction.get(xid);
        try {
            if (!autoAbort) {
                activeTransaction.remove(xid);
            }
        } finally {
            lock.unlock();
        }
        if (transaction.isAutoAborted()) return;
        lockMap.remove(xid);
        transactionManager.abort(xid);
    }

    /**
     * 释放一个entry的缓存引用
     *
     * @param versionEntry versionEntry
     */
    public void releaseEntry(VersionEntry versionEntry) {
        super.release(versionEntry.getUid());
    }

    @Override
    protected VersionEntry getForCache(long key) throws Exception {
        return VersionEntry.loadEntry(this, key);
    }

    @Override
    protected void releaseForCache(VersionEntry entry) {
        entry.remove();
    }

    public DataManager getDataManager() {
        return dataManager;
    }
}
